package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeProvinceOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.codehaus.plexus.util.StringUtils;

import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class DwsTradeProvinceOrderWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow().start(10020, 4, "dws_trade_province_order_window", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)){
                            collector.collect(JSON.parseObject(jsonStr));
                        }
                    }
                }
        );
        // jsonObjDS.print();
        // TODO 按照唯一键id进行分组
        // {"create_time":"2024-08-03 19:36:33","sku_num":"1","split_original_amount":"9899.0000","split_coupon_amount":"0.0",
        // "sku_id":"13","date_id":"2024-08-03","user_id":"5770","province_id":"9","sku_name":"联想（Lenovo） 拯救者Y9000P 2022 16英寸游戏笔记本电脑 i9-12900H RTX3060 钛晶灰",
        // id":"30030","order_id":"21544","split_activity_amount":"0.0","split_total_amount":"9899.0","ts":1654774593}
        KeyedStream<JSONObject, String> idKeyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getString("id"));
        // TODO 去重 （抵消 + 状态）
        SingleOutputStreamOperator<JSONObject> distinctDS = idKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    ValueState<JSONObject> lastVisitJsonState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("lastVisitJsonState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(60)).build());
                        lastVisitJsonState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject lastJsonObj = lastVisitJsonState.value();
                        if (lastJsonObj != null) {
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);
                            collector.collect(lastJsonObj);
                        }
                        lastVisitJsonState.update(jsonObject);
                        collector.collect(jsonObject);
                    }
                }
        );
        // TODO 类型转换jsonObj -> 实体类对象
        SingleOutputStreamOperator<TradeProvinceOrderBean> beanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean map(JSONObject jsonObject) throws Exception {
                        return TradeProvinceOrderBean.builder()
                                .orderDetailId(jsonObject.getString("order_id"))
                                .provinceId(jsonObject.getString("province_id"))
                                .orderAmount(jsonObject.getBigDecimal("split_total_amount"))
                                .orderIdSet(new HashSet<>(Collections.singleton(jsonObject.getString("order_id"))))
                                .ts(jsonObject.getLong("ts") * 1000)
                                .build();
                    }
                }
        );
        //beanDS.print();
        // TODO 设置水位线，提取时间字段
        SingleOutputStreamOperator<TradeProvinceOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeProvinceOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeProvinceOrderBean tradeProvinceOrderBean, long l) {
                                        return tradeProvinceOrderBean.getTs();
                                    }
                                }
                        )
        );
        // TODO 按照统计的维度 省份id 进行分组
        KeyedStream<TradeProvinceOrderBean, String> provinceIdKeyedDS = withWatermarkDS.keyBy(TradeProvinceOrderBean::getProvinceId);
        // TODO 开窗
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDS = provinceIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        // TODO 聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        return value1;
                    }
                },
                new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                        TradeProvinceOrderBean tradeProvinceOrderBean = iterable.iterator().next();
                        tradeProvinceOrderBean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        tradeProvinceOrderBean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        tradeProvinceOrderBean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        tradeProvinceOrderBean.setOrderCount((long) tradeProvinceOrderBean.getOrderIdSet().size());
                        collector.collect(tradeProvinceOrderBean);
                    }
                }
        );
        // reduceDS.print();
        SingleOutputStreamOperator<TradeProvinceOrderBean> witheBaseProvinceDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>() {
                    @Override
                    public String getRowKey(TradeProvinceOrderBean tradeProvinceOrderBean) {
                        return tradeProvinceOrderBean.getProvinceId();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }

                    @Override
                    public void addDims(TradeProvinceOrderBean tradeProvinceOrderBean, JSONObject jsonObj) {
                        tradeProvinceOrderBean.setProvinceName(jsonObj.getString("name"));
                    }
                },
                60, TimeUnit.SECONDS
        );
        witheBaseProvinceDS.print();
        // TODO 写入Doris
        witheBaseProvinceDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_province_order_window"));
    }

}
